package com.joven.internet.mq;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.joven.core.model.Person;
import com.joven.internet.constant.JovenInput;
import com.joven.internet.entity.Account;
import com.joven.internet.mapper.AccountMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * @author joven_wong
 * 2021/6/11 11:02
 * EnableBinding：该注解用来指定一个或多个定义了@Input或@Output注解的接口，以此实现对消息通道（Channel）的绑定
 * StreamListener：该注解主要定义在方法上，作用是将被修饰的方法注册为消息中间件上数据流的事件监听器，注解中的属性值对应了监听的消息通道名
 */
@Component
@Slf4j
public class RocketMqConsumer {
    @Autowired
    AccountMapper accountMapper;

    @StreamListener(JovenInput.INPUT)
    @Transactional(rollbackFor = Exception.class)
    public void input(Person person) {
        QueryWrapper<Account> wrapper = new QueryWrapper<>(Account.builder().name(person.getName()).build());
        Account account = accountMapper.selectOne(wrapper);
        if(account == null){
            throw new IllegalArgumentException("该用户不存在: " + person.getName());
        }
        account.setScore(account.getScore() + 1);
        accountMapper.update(account,null);
        System.out.println("success to update");
    }

    @StreamListener(JovenInput.DLQ_INPUT)
    @Transactional(rollbackFor = Exception.class)
    public void dlqInput(Person person) {
        log.error("test-score失败:{}", person);
    }
}
